package com.hexiang.consumer;

import io.openmessaging.Message;
import io.openmessaging.MessagingAccessPoint;
import io.openmessaging.OMS;
import io.openmessaging.OMSBuiltinKeys;
import io.openmessaging.consumer.PushConsumer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

@Slf4j
@Component
public class OmsPushConsumer {

    private MessagingAccessPoint messagingAccessPoint;

    private PushConsumer consumer;

    @PostConstruct
    public void init() {
        messagingAccessPoint = OMS.getMessagingAccessPoint("oms:rocketmq://localhost:9876/default:default");
        consumer = messagingAccessPoint.createPushConsumer(OMS.newKeyValue().put(OMSBuiltinKeys.CONSUMER_ID, "OMS_CONSUMER"));
        messagingAccessPoint.startup();
        consumer.attachQueue("OMS_HELLO_TOPIC", (message, context) -> {
            try {
                log.info("Received one message: {};messageId:{}", new String(message.getBody(byte[].class)), message.sysHeaders().getString(Message.BuiltinKeys.MESSAGE_ID));
                context.ack();
            } catch (Exception ex) {
                log.error("", ex);
            }

        });
        consumer.startup();
    }

    @PreDestroy
    public void shutdown() {
        consumer.shutdown();
        messagingAccessPoint.shutdown();
    }

}